package com.tjcd.project.library_base.push.mqtt;

import android.text.TextUtils;

import com.blankj.utilcode.util.LogUtils;
import com.blankj.utilcode.util.ToastUtils;
import com.jeremyliao.liveeventbus.LiveEventBus;
import com.tjcd.project.library_base.BaseApplication;
import com.tjcd.project.library_base.common.basicInfor.SugarSpConst;
import com.tjcd.project.library_base.common.basicInfor.SugarUrls;

import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.CompositeDisposable;
import io.reactivex.rxjava3.schedulers.Schedulers;


/**
 * @Author jiangKunKun
 * @CreateDate 2021/7/12
 * @Description MQTT 管理类，用于管理 MQTT 连接、订阅和消息处理
 * @SpittingGrooves
 */
public class MqttManager {
    /**
     * client 静态 MQTT 客户端实例
     */
    public static MqttAndroidClient mqttAndroidClient;
    /**
     * 服务器地址（协议+地址+端口号）  47.104.171.188:8080
     */
    public static String HOST = "tcp://39.105.89.75:1883";
    public static String HOST_DEBUG = "tcp://39.105.89.75:1883";
    /**
     * 用户名 和 密码
     * 防止任何人都可以连接
     */
    public static String USERNAME = "admin";
    public static String USERNAME_DEBUG = "admin";
    public static String PASSWORD = "jieoa2023.";
    public static String PASSWORD_DEBUG = "jieoa2023.";
    /**
     * 设备唯一标识
     */
    public static String CLIENT_ID = "Android_" + SugarSpConst.getMemberId() + "_" + System.currentTimeMillis();
    /**
     * mqtt是否连接
     */
    public static boolean isConnect = false;
    /**
     * 已订阅的
     */
    private static Set<String> subscribedTopics = new HashSet<>();

    /**
     * 静态单例实例
     */
    private static volatile MqttManager instance;
    /**
     * MQTT 连接选项
     */
    private MqttConnectOptions mMqttConnectOptions;
    /**
     * 用于管理 RxJava 订阅
     */
    private static CompositeDisposable disposables = new CompositeDisposable();
    /**
     * 最大订阅数量，避免内存泄漏
     */
    private static final int MAX_DISPOSABLES = 50;
    private MqttManager() {
        initMqtt();
    }

    /**
     * 获取 MqttManager 的单例实例
     *
     * @return MqttManager 实例
     */
    public static MqttManager getInstance() {
        if (instance == null) {
            synchronized (MqttManager.class) {
                if (instance == null) {
                    instance = new MqttManager();
                }
            }
        }
        return instance;
    }

    /**
     * 获取mqtt客户端
     *
     * @return
     */
    public static MqttAndroidClient getMqttClient() {
        if (null == mqttAndroidClient) {
            try {
                HOST = SugarUrls.DEBUG ? HOST_DEBUG : HOST;
                USERNAME = SugarUrls.DEBUG ? USERNAME_DEBUG : USERNAME;
                PASSWORD = SugarUrls.DEBUG ? PASSWORD_DEBUG : PASSWORD;
                LogUtils.e("mqtt连接的host为: " + HOST + "\nclentId为:" + CLIENT_ID);
                mqttAndroidClient = new MqttAndroidClient(BaseApplication.getInstance(), HOST, CLIENT_ID);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return mqttAndroidClient;
    }

    /**
     * 初始化 MQTT 连接
     */
    private void initMqtt() {
        try {
            // 获取 MQTT 客户端实例
            mqttAndroidClient = getMqttClient();
            // 设置回调
            mqttAndroidClient.setCallback(mqttCallback);
            // mqtt设置
            mMqttConnectOptions = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            mMqttConnectOptions.setCleanSession(true);
            // 设置连接的用户名
            mMqttConnectOptions.setUserName(USERNAME);
            // 设置连接的密码
            mMqttConnectOptions.setPassword(PASSWORD.toCharArray());
            // 设置超时时间 单位为秒
            mMqttConnectOptions.setConnectionTimeout(10);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            mMqttConnectOptions.setKeepAliveInterval(27);
            // 设置自动断线重连
            mMqttConnectOptions.setAutomaticReconnect(true);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 连接监听
     */
    private IMqttActionListener iMqttActionListener = new IMqttActionListener() {

        @Override
        public void onSuccess(IMqttToken asyncActionToken) {
            LogUtils.e("mqtt连接服务器成功");
            MqttManager.isConnect = true;
//            EventBus.getDefault().post(new NetConnectMessage());
        }

        @Override
        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
            LogUtils.e("mqtt连接服务器失败");
            MqttManager.isConnect = false;
            LiveEventBus.get("mqtt").post(new MqttConnectEvent("fail"));
            reConnectMqtt();
        }


    };

    /**
     * 订阅回调
     */
    private MqttCallback mqttCallback = new MqttCallbackExtended() {

        @Override
        public void messageArrived(String topic, MqttMessage msgStr) {
            try {
                String payload = new String(msgStr.getPayload());
                LogUtils.e("mqtt接受到的数据: " + payload + "\ntopic: " + topic);
                if (isJson(payload)) {
                    LogUtils.json(payload);
                } else {
                    LogUtils.e(payload);
                    ToastUtils.showLong("数据结构错误");
                    return;
                }
                String msg = new String(msgStr.getPayload());
                switch (topic) { // 订阅的频道
                    case MqttConfig.ACS_STATE_CHANGE :
//                    AccessBean accessBean = GsonUtils.GsonToBean(msg, AccessBean.class);
//                    EventBus.getDefault().post(accessBean);
                        break;
                    default:break;
                }
            } catch (Exception e) {
                e.printStackTrace();
                LogUtils.e("mqtt下发的数据异常: " + Arrays.toString(msgStr.getPayload()) + "\ntopic: " + topic);
            }

        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken arg0) {
            // 消息发送完成回调
        }

        @Override
        public void connectionLost(Throwable arg0) {
            LogUtils.e("mqtt断开服务器连接");
            MqttManager.isConnect = false;
            LiveEventBus.get("mqtt").post(new MqttConnectEvent("error"));
            reConnectMqtt();
        }

        /**
         * MQTT 连接完成（重连时 reconnect 为 true）
         *
         * @param reconnect 是否为重连
         * @param serverURI 服务器 URI
         */
        @Override
        public void connectComplete(boolean reconnect, String serverURI) {
            LogUtils.e("mqtt连接服务器完成");
            MqttManager.isConnect = true;
            LiveEventBus.get("mqtt").post(new MqttConnectEvent("suc"));
        }
    };

    /**
     * 连接mqtt
     */
    public void connectMqtt() {
        try {
            if (null != mqttAndroidClient && !mqttAndroidClient.isConnected()) {
                LogUtils.e("去连接...");
                mqttAndroidClient.connect(mMqttConnectOptions, null, iMqttActionListener);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 重连mqtt
     */
    private void reConnectMqtt() {
        if (disposables.size() >= MAX_DISPOSABLES) {
            LogUtils.e("达到最大订阅数量，清除所有订阅并重新开始重连");
            disposables.clear();
        }

        disposables.add(Observable.interval(0, 5, TimeUnit.SECONDS)
                // 使用 flatMap 和 retryWhen 实现指数退避算法，避免频繁尝试连接导致资源浪费
                .flatMap(attempt -> Observable.just(attempt)
                        .doOnNext(a -> {
                            try {
                                if (null != mqttAndroidClient && !mqttAndroidClient.isConnected()) {
                                    LogUtils.e("尝试第 " + (a + 1) + " 次重连...");
                                    mqttAndroidClient.connect(mMqttConnectOptions, null, iMqttActionListener);
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        })
                        .retryWhen(errors -> errors.flatMap(error -> Observable.timer(5, TimeUnit.SECONDS)))
                )
                .subscribeOn(Schedulers.io()) // 避免上面的阻塞
                .subscribe(
                        attempt -> LogUtils.e("重连尝试: " + (attempt + 1)),
                        throwable -> LogUtils.e("重连时发生错误: " + throwable.getMessage())
                ));
    }


    /**
     * 关闭资源
     */
    public static void unregisterResources() {
        if (null != getMqttClient()) {
            mqttAndroidClient.unregisterResources();
            mqttAndroidClient.close();
        }
    }

    /**
     * 断开 MQTT 连接
     */
    public static void disconnectMqtt() {
        if (null != getMqttClient()) {
            try {
                if (mqttAndroidClient != null) {
                    mqttAndroidClient.disconnect();
                    mqttAndroidClient = null;
                }
                disposables.clear(); // 清除所有订阅，避免内存泄漏
            } catch (Exception e) {
            } finally {
                mqttAndroidClient = null;
            }

        }
    }


    /**
     * 订阅主题
     *
     * @param topic 主题
     */
    public static void subscribe(String topic) {
        subscribe(topic, 0);
    }

    /**
     * 订阅主题
     *
     * @param topic 主题
     * @param qos   服务质量（0、1 或 2）
     */
    public static void subscribe(String topic, int qos) {
        try {
            if (null != getMqttClient()) {
                LogUtils.e("mqtt订阅主题: " + topic);
                getMqttClient().subscribe(topic, qos);
                subscribedTopics.add(topic);
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 取消订阅
     *
     * @param topic 主题
     */
    public static void unSubscribe(String topic) {
        try {
            if (null != getMqttClient()) {
                LogUtils.e("mqtt取消订阅主题: " + topic);
                getMqttClient().unsubscribe(topic);
                subscribedTopics.remove(topic);
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布 （模拟其他客户端发布消息）
     * @param topic 发布消息的主题
     * @param message 信息
     */
    public static void publish(String topic, String message) {
        publish(topic, message , 2, false);
    }

    /**
     * 发布 （模拟其他客户端发布消息）
     * @param topic 发布消息的主题
     * @param message 信息
     * @param qos 提供消息的服务质量，可传0、1或2
     * @param retained 是否在服务器保留断开连接后的最后一条消息
     */
    public static void publish( String topic, String message, int qos, boolean retained) {
        LogUtils.e("mqtt推送 message: " + message + "\ntopic: " + topic);
        try {
            //参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息
            if (getMqttClient() != null) {
                getMqttClient().publish(topic, message.getBytes(), qos, retained);
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    /**
     * 刷新 MQTT 连接
     */
    public static void refreshMqtt() {
        unregisterResources();
        disconnectMqtt();
        isConnect = false;
        // 重新初始化 MQTT 客户端
        getInstance().initMqtt();
        // 重新连接 MQTT
        getInstance().connectMqtt();
    }

    /**
     * 判断 是否是json
     * @param json
     * @return
     */
    private boolean isJson(String json) {
        if (TextUtils.isEmpty(json)) {
            return false;
        }
        try {
            new JSONObject(json);
        } catch (JSONException ex) {
            try {
                new JSONArray(json);
            } catch (JSONException ex1) {
                return false;
            }
        }
        return true;
    }
}
